package redisclient;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RedisClient implements Runnable {

    private static Logger logger = LoggerFactory.getLogger(RedisClient.class);

    private final String redisserver;
    private final int redisport;

    private final Selector selector;
    private final ExecutorService executorService;

    public RedisClient(String redisserver, int redisport) {
        this.redisserver = redisserver;
        this.redisport = redisport;

        try {
            this.selector = Selector.open();
            this.executorService = Executors.newSingleThreadExecutor();
            executorService.submit(this);
        } catch (IOException e) {
            throw new RedisException(e);
        }
    }

    @Override
    public void run() {
        while (true) {
            try {

                logger.info("before select");
                int count = selector.select();
                logger.info("select, count={}", count);

                if (count > 0) {
                    Iterator<SelectionKey> selectionKeys = selector.selectedKeys().iterator();
                    while (selectionKeys.hasNext()) {
                        SelectionKey selectionKey = selectionKeys.next();
                        selectionKeys.remove();

                        if (selectionKey.isConnectable()) {
                            logger.info("isConnectable");
                        }

                        if (selectionKey.isWritable()) {
                            logger.info("isWritable");
                            RedisConnection channel = (RedisConnection) selectionKey.attachment();
                            channel.write();
                        }

                        if (selectionKey.isReadable()) {
                            logger.info("isReadable");
                            RedisConnection channel = (RedisConnection) selectionKey.attachment();
                            channel.read();
                        }
                    }
                }

                Thread.sleep(1000 * 3);
            } catch (IOException e) {
                throw new RedisException(e);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public RedisConnection connection() {
        try {
            SocketChannel socketChannel = SocketChannel.open();
            socketChannel.connect(new InetSocketAddress(this.redisserver, this.redisport));

            socketChannel.configureBlocking(false);
            logger.info("before register");
            // 不执行wakeup会被锁住
            // https://stackoverflow.com/questions/3189153/java-non-blocking-io-selector-causing-channel-register-to-block
            selector.wakeup();
            SelectionKey selectionKey = socketChannel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ);
            logger.info("after register");

            RedisConnection redisConnection = new RedisConnection(this.selector, selectionKey, socketChannel);
            selectionKey.attach(redisConnection);

            return redisConnection;
        } catch (IOException e) {
            throw new RedisException(e);
        }
    }


    public void shutdown() {

        this.executorService.shutdownNow();

        try {
            this.selector.close();
        } catch (IOException e) {
            logger.error(e.getMessage(), e);
        }

    }

}
